package ex.other;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class StreamingJob {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.
                getExecutionEnvironment();
        DataStream<String> text = env.socketTextStream("127.0.0.1", 9999, "\n");
        DataStream<WordWithCount> windowCounts = text.flatMap(new FlatMapFunction<String, WordWithCount>() {
            @Override
            public void flatMap(String value, Collector<WordWithCount> out) throws Exception {
                for (String word : value.split("\\s")) {
                    out.collect(new WordWithCount(word, 1L));
                }
            }
        })
        .keyBy("word")
//        .timeWindow(Time.seconds(5), Time.seconds(1))
        .window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5))) //娣诲姞婊戝姩绐楀彛锛岄噰鐢ㄥ鐞嗘椂闂磋繘琛屽鐞�
        .reduce(new ReduceFunction<WordWithCount>() {
            @Override
            public WordWithCount reduce(WordWithCount a, WordWithCount b) throws Exception {
                return new WordWithCount(a.getWord(), a.getCount() + b.getCount());
            }
        });
        windowCounts.print().setParallelism(1);
        env.execute("sicjet Window WordCount");

    }
    public static class WordWithCount {
        private String word;
        private Long count;

        @Override
        public String toString() {
            return "WordWithCount{" +
                    "word='" + word + '\'' +
                    ", count=" + count +
                    '}';
        }

        public WordWithCount() {
        }

        public WordWithCount(String word, long l) {
            this.word=word;
            this.count=l;
        }

        public String getWord() {
            return word;
        }

        public void setWord(String word) {
            this.word = word;
        }

        public Long getCount() {
            return count;
        }

        public void setCount(Long count) {
            this.count = count;
        }
    }
/*



C:\Users\Administrator>nc -l -p 9999
aa bb cc
dd ee ff

C:\Users\Administrator>nc -l -p 9999
tt hh bb cccc cccc aa z

WordWithCount{word='hh', count=1}
WordWithCount{word='tt', count=1}
WordWithCount{word='bb', count=1}
WordWithCount{word='aa', count=1}
WordWithCount{word='z', count=1}
WordWithCount{word='cccc', count=2}
WordWithCount{word='hh', count=1}
WordWithCount{word='bb', count=1}
WordWithCount{word='z', count=1}
WordWithCount{word='cccc', count=2}
WordWithCount{word='aa', count=1}
WordWithCount{word='tt', count=1}
* */
}
